package top.shiyiri.controller;

import com.alibaba.fastjson.JSONObject;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.apache.kafka.common.serialization.StringSerializer;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Controller;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.ResponseBody;
import top.shiyiri.bean.Advert;
import top.shiyiri.service.AdvertService;

import java.util.List;
import java.util.Properties;
import java.util.concurrent.Future;

@Controller
public class AdvertController {

    @Autowired
    private AdvertService advertService;

    @RequestMapping("/findAllAdverts")
    @ResponseBody
    public String findAllAdverts(){
        List<Advert> list = advertService.findAllAdverts();
        return JSONObject.toJSONString(list);
    }

    @RequestMapping("/clickAdvert")
    @ResponseBody
    public String clickAdvert(String id){
        //配置信息
        Properties pro = new Properties();
        //kafka服务器地址
        pro.put("bootstrap.servers", "master:9092,slave1:9092,slave2:9092");
        //设置数据key和value的序列化处理类
        pro.put("key.serializer", StringSerializer.class);
        pro.put("value.serializer", StringSerializer.class);

        //创建生产者实例
        KafkaProducer<String, String> producer = new KafkaProducer<>(pro);
        ProducerRecord<String, String> record = new ProducerRecord<>("test1", id, "click");

        //发送记录
        Future<RecordMetadata> send = producer.send(record);

        String result = send .isDone() + "";
        producer.close();
        return result;
    }
}


